package com.atguigu.flink.splitAndUnionStream;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Optional;

public class FullOuterJoinCoProcessFunction extends CoProcessFunction<
        Tuple2<String, Integer>,
        Tuple2<String, Integer>,
        Tuple2<String, Tuple2<Optional<Integer>, Optional<Integer>>>> {

    // 状态描述符，用于存储左流和右流的数据
    private ValueStateDescriptor<Tuple2<String, Integer>> leftStateDescriptor = new ValueStateDescriptor<>(
            "leftState", // 状态名称
            TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {})); // 状态类型

    private ValueStateDescriptor<Tuple2<String, Integer>> rightStateDescriptor = new ValueStateDescriptor<>(
            "rightState", // 状态名称
            TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {})); // 状态类型

    private transient ValueState<Tuple2<String, Integer>> leftState;
    private transient ValueState<Tuple2<String, Integer>> rightState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化状态
        leftState = getRuntimeContext().getState(leftStateDescriptor);
        rightState = getRuntimeContext().getState(rightStateDescriptor);
    }

    @Override
    public void processElement1(Tuple2<String, Integer> left, Context ctx, Collector<Tuple2<String, Tuple2<Optional<Integer>, Optional<Integer>>>> out) throws Exception {
        // 检查右流中是否有匹配的元素
        Tuple2<String, Integer> rightValue = rightState.value();
        if (rightValue != null) {
            // 如果右流中有匹配的元素，则输出并清除右流的状态
            out.collect(new Tuple2<>(left.f0, new Tuple2<>(Optional.of(left.f1), Optional.of(rightValue.f1))));
            rightState.clear();
        } else {
            // 如果右流中没有匹配的元素，则将左流的数据存储到状态中
            leftState.update(left);
        }
    }

    @Override
    public void processElement2(Tuple2<String, Integer> right, Context ctx, Collector<Tuple2<String, Tuple2<Optional<Integer>, Optional<Integer>>>> out) throws Exception {
        // 检查左流中是否有匹配的元素
        Tuple2<String, Integer> leftValue = leftState.value();
        if (leftValue != null) {
            // 如果左流中有匹配的元素，则输出并清除左流的状态
            out.collect(new Tuple2<>(right.f0, new Tuple2<>(Optional.of(leftValue.f1), Optional.of(right.f1))));
            leftState.clear();
        } else {
            // 如果左流中没有匹配的元素，则将右流的数据存储到状态中
            rightState.update(right);
        }
    }

}

// 在你的Flink程序中使用FullOuterJoinCoProcessFunction

